package worker

import (
	"bufio"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"mapreduce/common"
	"os"
	"strconv"
)

var node common.Node

func init() {
	node.Addr = "localhost:8022"
	node.Id = 2
}

func ReceiveTask(job common.Job, taskType string) map[int]string {
	switch taskType {
	case common.Map_Phase:
		return doMap(job)
	case common.Reduce_Phase:
		return doReduce(job)
	}
	return nil
}

func doMap(job common.Job) map[int]string {
	reduceNumberKeyFilePathMap := make(map[int]string)
	fmt.Println("Map编号：" + strconv.Itoa(job.TaskNumber) + "文件:" + job.TaskFilePathSlc[0] + "打开")
	filePath := job.TaskFilePathSlc[0]
	filePtr, openErr := os.OpenFile(filePath, os.O_RDONLY, 0666)
	defer filePtr.Close()
	if openErr != nil {
		fmt.Println(job.TaskFilePathSlc[0] + "文件打开失败！" + openErr.Error())
	}
	reader := bufio.NewReader(filePtr)
	var lineNum = 0
	var mapKVResultSlc []common.KeyValue
	for {
		lineSlc, _, readErr := reader.ReadLine()
		if readErr == io.EOF {
			fmt.Println("Map编号：" + strconv.Itoa(job.TaskNumber) + "   " + job.TaskFilePathSlc[0] + "文件读取完毕")
			break
		} else if readErr != nil {
			fmt.Println(job.TaskFilePathSlc[0] + "文件读取失败" + readErr.Error())
			break
		}
		lineNum++
		mapKVResultSlc = append(mapKVResultSlc, job.MapFunc(strconv.Itoa(lineNum), string(lineSlc))...)
	}
	//把Map计算的结果使用json的形式写入文件,按照Reduce的数量生成reduce个文件
	if len(mapKVResultSlc) > 0 {
		reduceNumberKeyFilePtrMap := make(map[int]*os.File)
		reduceNumberKeyEncoderPtrMap := make(map[int]*json.Encoder)
		for i := 0; i < job.ReduceNum; i++ {
			mapFilePath := common.GetMapFilePath(job.JobName, job.TaskNumber, i)
			reduceNumberKeyFilePathMap[i] = mapFilePath
			filePtr, openErr := os.OpenFile(mapFilePath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0666)
			if openErr != nil {
				fmt.Println(mapFilePath + "文件创建失败,原因:" + openErr.Error())
			}
			reduceNumberKeyEncoderPtrMap[i] = json.NewEncoder(filePtr)
			reduceNumberKeyFilePtrMap[i] = filePtr
		}
		for _, kvResult := range mapKVResultSlc {
			reduceIndex := common.ParseReduceHashIndex(kvResult.Key, job.ReduceNum)
			encodeErr := reduceNumberKeyEncoderPtrMap[reduceIndex].Encode(kvResult)
			if encodeErr != nil {
				fmt.Printf("数据：%v,写入%f文件失败！异常：%s\n", kvResult, reduceNumberKeyFilePtrMap[reduceIndex].Name(), encodeErr.Error())
			}
		}
		fmt.Printf("Map编号：%v,生成的文件：%v\n", job.TaskNumber)
		//释放文件对象
		defer func() {
			for _, filePtr := range reduceNumberKeyFilePtrMap {
				fmt.Println("关闭" + filePtr.Name() + "文件")
				filePtr.Close()
			}
		}()
	}
	return reduceNumberKeyFilePathMap
}

func doReduce(job common.Job) map[int]string {
	fmt.Printf("Reduce编号：%v,Reduce的文件：%v\n", job.TaskNumber, job.TaskFilePathSlc)
	reduceMap := make(map[string][]string)
	// 返回类型是map，有返回变量名时还需要make该返回变量名，不然报nil map
	reduceFileMap := make(map[int]string)
	for _, reduceFilePath := range job.TaskFilePathSlc {
		filePtr, openErr := os.OpenFile(reduceFilePath, os.O_RDONLY, 0666)
		if openErr != nil {
			fmt.Println("文件读取失败：" + reduceFilePath + "，原因:" + openErr.Error())
			filePtr.Close()
			break
		}
		fmt.Printf("读取%v文件\n", reduceFilePath)
		decoder := json.NewDecoder(filePtr)
		var keyValue common.KeyValue
		for decoder.More() {
			decoderErr := decoder.Decode(&keyValue)
			if decoderErr != nil {
				fmt.Printf("读取的数据无法解析%v\n", keyValue)
				break
			}
			if _, ok := reduceMap[keyValue.Key]; ok {
				valueSlc := reduceMap[keyValue.Key]
				valueSlc = append(valueSlc, keyValue.Value)
				reduceMap[keyValue.Key] = valueSlc
			} else {
				valueSlc := make([]string, 1)
				valueSlc[0] = keyValue.Value
				reduceMap[keyValue.Key] = valueSlc
			}
			//fmt.Printf("Reduce读取到的数据%v\n",keyValue)
		}
		filePtr.Close()
	}

	reduceFilePath := common.GetReduceFileName(job.JobName, job.TaskNumber)
	reduceFilePtr, openErr := os.OpenFile(reduceFilePath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0666)
	if openErr != nil {
		fmt.Println("reduce文件创建失败" + reduceFilePath + "，异常：" + openErr.Error())
	}
	encoder := json.NewEncoder(reduceFilePtr)
	defer reduceFilePtr.Close()
	//Reduce生成的文件路径
	reduceFileMap[job.TaskNumber] = reduceFilePath
	if openErr != nil {
		fmt.Println("创建reduceFile失败+" + reduceFilePath + " error:" + openErr.Error())
	}
	for key, valueSlc := range reduceMap {
		reduceResult := job.ReduceFunc(key, valueSlc)
		encoderErr := encoder.Encode(reduceResult)
		if encoderErr != nil {
			log.Fatalf("写入失败%v", encoderErr)
		}
	}
	return reduceFileMap
}
